/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.messaging.kafka.consumer;

import org.ipig.commons.service.ProcessorService;
import org.ipig.constants.ResultStatus;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecords;

/**
 * GenericKafkaConsumer
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: GenericKafkaConsumer.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
@Data
public class GenericKafkaConsumer extends AbstractKafkaConsumer {
    /**
     * 处理器服务
     */
    private ProcessorService<ConsumerRecords<String, String>, ResultStatus> processorService;

    @Override
    public ResultStatus process(ConsumerRecords<String, String> records) {
        ResultStatus resultStatus = ResultStatus.SUCCESS;
        try {
            if (processorService != null) {
                resultStatus = processorService.process(records);
            } else {
                log.warn("no available Processor!");
                resultStatus = ResultStatus.FAILURE;
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error(ExceptionUtils.getStackTrace(ex));
            resultStatus = ResultStatus.FAILURE;
        }
        return resultStatus;
    }
}